/**
 * 
 */
package com.battcn.config;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.config.SimpleRabbitListenerContainerFactory;
import org.springframework.amqp.rabbit.connection.CachingConnectionFactory;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.rabbit.support.CorrelationData;
import org.springframework.amqp.support.converter.Jackson2JsonMessageConverter;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.amqp.SimpleRabbitListenerContainerFactoryConfigurer;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.env.Environment;

import java.util.HashMap;
import java.util.Map;

/**
 * @author yanglei 2018年1月15日
 */
@Configuration
public class RabbitmqConfig {

	private final static Logger log = LoggerFactory.getLogger("mqLog");

	@Autowired
	private Environment env;

	@Autowired
	private CachingConnectionFactory connectionFactory;

	@Autowired
	private SimpleRabbitListenerContainerFactoryConfigurer factoryConfigurer;

	/**
	 * 延迟队列配置
	 * 
	 * producer 发送消息时设置消息的 expiration 属性，然后将消息发送到延时exchange，而后在 expiration
	 * 时间后消息将被转发到x-dead-letter-exchange 指定的 exchange 去，而后被正常消费，实现了延时消息功能。
	 **/
                              
	@Bean(name = "registerDelayQueue")
	public Queue registerDelayQueue() {
		Map<String, Object> params = new HashMap<>();
		params.put("x-dead-letter-exchange", env.getProperty("register.exchange.name"));
		params.put("x-dead-letter-routing-key", "all");
		// 队列名称 队列持久化 排他性 自动删除

		// 将exclusive置为True，设置了排外为true的队列只可以在本次的连接中被访问，也就是说在当前连接创建多少个channel访问都没有关系
		// 排外的queue在当前连接被断开的时候会自动消失（清除）无论是否设置了持久化
		// autodelete 也就是说queue会清理自己 但是是在最后一个connection断开的时候
		return new Queue(env.getProperty("register.delay.queue.name"), true, false, false, params);
	}

	@Bean
	public DirectExchange registerDelayExchange() {
		// 声明Direct模式交换器的名字
		return new DirectExchange(env.getProperty("register.delay.exchange.name"));
	}

	// 绑定对列与交换器
	@Bean
	public Binding registerDelayBinding() {
		return BindingBuilder.bind(registerDelayQueue()).to(registerDelayExchange()).with("");
	}

	/** 延迟队列配置 **/

	/** 指标消费队列配置 **/

	@Bean
	public TopicExchange registerTopicExchange() {
		//
		return new TopicExchange(env.getProperty("register.exchange.name"));
	}
	
	
	//topic模式 根据routingKey匹配  假如监听绑定配置test.*，则生产者匹配rabbitTemplate.convertAndSend(env.getProperty("register.exchange.name"),"test.a", message); 
	@Bean
	public Binding registerBinding() {
		return BindingBuilder.bind(registerQueue()).to(registerTopicExchange()).with("all");
	}

	@Bean(name = "registerQueue")
	public Queue registerQueue() {
		return new Queue(env.getProperty("register.queue.name"), true);
	}

	/** 指标消费队列配置 **/

	/**
	 * 单一消费者
	 * 
	 * @return
	 */
	@Bean(name = "singleListenerContainer")
	public SimpleRabbitListenerContainerFactory listenerContainer() {
		SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
		factory.setConnectionFactory(connectionFactory);
		factory.setMessageConverter(new Jackson2JsonMessageConverter());
		factory.setConcurrentConsumers(1);
		factory.setMaxConcurrentConsumers(1);
		factory.setPrefetchCount(1);
		factory.setTxSize(1);
		return factory;
	}

	/**
	 * 多个消费者
	 * 
	 * @return
	 */
	@Bean(name = "multiListenerContainer")
	public SimpleRabbitListenerContainerFactory multiListenerContainer() {
		SimpleRabbitListenerContainerFactory factory = new SimpleRabbitListenerContainerFactory();
		factoryConfigurer.configure(factory, connectionFactory);
		factory.setMessageConverter(new Jackson2JsonMessageConverter());
		factory.setAcknowledgeMode(AcknowledgeMode.NONE);
		return factory;
	}

	@Bean
	public RabbitTemplate rabbitTemplate() {
		connectionFactory.setPublisherConfirms(true);
		connectionFactory.setPublisherReturns(true);
		RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
		rabbitTemplate.setMandatory(true);
		rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
			@Override
			public void confirm(CorrelationData correlationData, boolean ack, String cause) {
				log.info("消息发送成功:correlationData({}),ack({}),cause({})", correlationData, ack, cause);
			}
		});
		rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
			@Override
			public void returnedMessage(Message message, int replyCode, String replyText, String exchange,
					String routingKey) {
				log.info("消息丢失:exchange({}),route({}),replyCode({}),replyText({}),message:{}", exchange, routingKey,
						replyCode, replyText, message);
			}
		});
		return rabbitTemplate;
	}

}
